package cn.juque.jdfsystem.task;

import cn.hutool.core.collection.CollUtil;
import cn.juque.common.disruptor.DisruptorHandler;
import cn.juque.jdfsystem.disruptor.bo.SendMsgBodyBO;
import cn.juque.jdfsystem.disruptor.event.SendMsgDisruptorEventServiceImpl;
import cn.juque.jdfsystem.entity.RegisterInfo;
import cn.juque.jdfsystem.enums.HttpMessageTypeEnum;
import cn.juque.jdfsystem.mapper.RegisterInfoMapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author juque
 * @version 1.0.0
 * <ul>
 *     <li>RegisterInfoUpdateTask</li>
 * </ul>
 * @date 2023-01-11 09:24:42
 **/
@Service("registerInfoUpdateTask")
public class RegisterInfoUpdateTask extends AbstractTask {

    @Resource
    private RegisterInfoMapper registerInfoMapper;

    @Resource
    private DisruptorHandler disruptorHandler;

    /**
     * 任务逻辑
     *
     * @param args 参数
     */
    @Override
    public void execute(String[] args) {
        if (Boolean.FALSE.equals(super.checkExecuteNode())) {
            return;
        }
        RegisterInfo query = new RegisterInfo();
        // 更新缓存，并同步数据库
        List<RegisterInfo> sourceList = this.registerInfoMapper.selectList(new QueryWrapper<>(query));
        int count = this.add(sourceList);
        count = count + this.update(sourceList);
        count = count + this.delete(sourceList);
        if (count > 0) {
            // 通知其他节点
            this.sendMessage();
        }
    }

    /**
     * 设置随机休眠等待最大时长 单位：秒
     *
     * @return 秒
     */
    @Override
    protected Integer initSleepTimes() {
        return null;
    }

    @Override
    protected String getTaskName() {
        return "RegisterInfoUpdateTask";
    }

    /**
     * 添加到缓存
     *
     * @param sourceList 数据库数据
     * @return 添加数量
     */
    private Integer add(List<RegisterInfo> sourceList) {
        if (CollUtil.isEmpty(sourceList)) {
            return 0;
        }
        Map<String, RegisterInfo> registerMap = this.registerInfoCache.getAllRegister();
        List<RegisterInfo> list = sourceList.stream()
                .filter(f -> !registerMap.containsKey(f.getNodeName())).collect(Collectors.toList());
        list.forEach(f -> this.registerInfoCache.update(f));
        return list.size();
    }

    /**
     * 同步变更到数据库
     *
     * @param sourceList 数据库数据
     * @return 添加数量
     */
    private Integer update(List<RegisterInfo> sourceList) {
        if (CollUtil.isEmpty(sourceList)) {
            return 0;
        }
        Map<String, RegisterInfo> registerMap = this.registerInfoCache.getAllRegister();
        List<RegisterInfo> list = sourceList.stream()
                .filter(f -> registerMap.containsKey(f.getNodeName()))
                .filter(f -> !f.getExecuteStatus().equals(registerMap.get(f.getNodeName()).getExecuteStatus()))
                .map(m -> registerMap.get(m.getNodeName())).collect(Collectors.toList());
        list.forEach(f -> this.registerInfoMapper.updateById(f));
        return list.size();
    }

    /**
     * 删除缓存
     *
     * @param sourceList 数据库数据
     * @return 添加数量
     */
    private Integer delete(List<RegisterInfo> sourceList) {
        Map<String, RegisterInfo> sourceMap = sourceList.stream().collect(Collectors.toMap(RegisterInfo::getNodeName, t -> t, (k1, k2) -> k2));
        Map<String, RegisterInfo> registerMap = this.registerInfoCache.getAllRegister();
        List<RegisterInfo> list = registerMap.values().stream()
                .filter(f -> !sourceMap.containsKey(f.getNodeName())).collect(Collectors.toList());
        list.forEach(f -> this.registerInfoCache.remove(f));
        return list.size();
    }

    /**
     * 发送消息
     */
    private void sendMessage(){
        SendMsgBodyBO<Boolean> sendMsgBodyBO = new SendMsgBodyBO<>();
        sendMsgBodyBO.setBusinessEventService(SendMsgDisruptorEventServiceImpl.BEAN);
        sendMsgBodyBO.setMessageTypeEnum(HttpMessageTypeEnum.REGISTER_CHANGE);
        this.disruptorHandler.publish(sendMsgBodyBO);
    }
}
